import com.sun.net.httpserver.Headers;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import com.sun.net.httpserver.spi.HttpServerProvider;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.compat.java8.JFunction0$mcB$sp;

import java.io.*;

import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.concurrent.Executor;


public class Request_Source extends RichSourceFunction<String> {


    private static final Logger LOGGER = LoggerFactory.getLogger(Request_Source.class);
    private static  HttpServerProvider provider;
    private static HttpServer httpserver;
    private static  Integer port;
    private static Integer connNumber;
    private volatile String response=""; //使用volatile完成变量在多个线程之间的通讯任务。
    private volatile boolean flag=false;
    /**
     *
     * @param port 本地端口，如果这里设置为9999.则业务系统需要访问的地址为: "本机IP地址:9999"
     * @param connNumber 最大同时接收端口数量
     * @throws IOException
     */
    public Request_Source(Integer port,Integer connNumber){
        this.port=port;
        this.connNumber=connNumber;
    }


    @Override
    public void run(SourceContext ctx) throws Exception {

        new Thread(new Runnable() {
            @Override
            public void run() {
                HttpServerProvider provider = HttpServerProvider.provider();
                HttpServer httpserver = null;//监听端口19017,能同时接受100个请求
                try {
                    httpserver = provider.createHttpServer(new InetSocketAddress(port), connNumber);
                } catch (IOException e) {
                    e.printStackTrace();
                }
                //监听的地址，如果是用/，表示向该端口发送的所有地址都会被监听

                httpserver.createContext("/", new HttpHandler() {
                    @Override
                    public void handle(HttpExchange httpExchange) throws IOException {
                        System.out.println("------开始处理------");
                        //设置服务端响应的编码格式，否则在客户端收到的可能是乱码
                        Headers responseHeaders = httpExchange.getResponseHeaders();
                        responseHeaders.set("Content-Type", "json;charset=utf-8");

                        //获得请求方法
                        String requestMethod = httpExchange.getRequestMethod();
                        if (requestMethod.equalsIgnoreCase("GET")) {//客户端的请求是get方法
                            //在这里通过httpExchange获取客户端发送过来的消息
                            //获取用户URI请求地址，需要再打开
                            //URI url = httpExchange.getRequestURI();
                            //这里不用手动关闭流，获取完请求体信息后流自动会关闭
                            InputStream requestBody = httpExchange.getRequestBody();
                            response=readStream(requestBody);
                            flag=true;
                            String response = "请求返回";

                            httpExchange.sendResponseHeaders(HttpURLConnection.HTTP_OK, response.getBytes("UTF-8").length);

                            OutputStream responseBody = httpExchange.getResponseBody();
                            OutputStreamWriter writer = new OutputStreamWriter(responseBody, "UTF-8");
                            writer.write(response);
                            writer.close();
                            responseBody.close();


                        }
                    }
                });
                //设置请求的线程执行器, 设置为 null 表示使用默认的执行器，封装了http请求和响应的所有数据操作
                httpserver.setExecutor(null);
                httpserver.start();
                System.out.println("server started");

            }
        }).start();
        while (true){
            while (flag){
                ctx.collect(response);
                flag=false;
            }
        }
    }

    @Override
    public void cancel() {

    }

    /**
     * 读取 InputStream 到 String字符串中
     */
    public static String readStream(InputStream in) {
        try {
            //<1>创建字节数组输出流，用来输出读取到的内容
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            //<2>创建缓存大小
            byte[] buffer = new byte[1024]; // 1KB
            //每次读取到内容的长度
            int len = -1;
            //<3>开始读取输入流中的内容
            while ((len = in.read(buffer)) != -1) { //当等于-1说明没有数据可以读取了
                baos.write(buffer, 0, len);   //把读取到的内容写到输出流中
            }
            //<4> 把字节数组转换为字符串
            String content = baos.toString();
            //<5>关闭输入流和输出流
            in.close();
            baos.close();
            //<6>返回字符串结果
            return content;
        } catch (Exception e) {
            e.printStackTrace();
            return  e.getMessage();
        }
    }

}